package com.example.mqtt_client.client;

import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.mqtt_client.entity.ConsumeUploadQos;
import com.example.mqtt_client.entity.Cron;
import com.example.mqtt_client.entity.User;
import com.example.mqtt_client.mapper.ConsumeUploadQosMapper;
import com.example.mqtt_client.mapper.CronMapper;
import com.example.mqtt_client.mapper.UserMapper;
import com.example.mqtt_client.singleton.Singleton;
import com.example.mqtt_client.util.DynamicScheduleTask;
import io.moquette.broker.Server;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.stream.Collectors;

import static java.nio.charset.StandardCharsets.UTF_8;

@Configuration
@EnableScheduling
public class MessagePushClient extends DynamicScheduleTask {

    @Autowired
    private ConsumeUploadQosMapper consumeUploadQosMapper;
    @Autowired
    private UserMapper userMapper;
    @Autowired
    private CronMapper cronMapper;


    @Override
    @Transactional
    public void scheduleWork() {
//        System.out.println(MessagePushClient.class.getSimpleName() + System.currentTimeMillis()/1000);
        QueryWrapper<ConsumeUploadQos> push = new QueryWrapper<ConsumeUploadQos>().eq("is_push", 0).last("limit 2");
        List<ConsumeUploadQos> consumeUploadQos = consumeUploadQosMapper.selectList(push);

        List<Long> ids = consumeUploadQos.stream().map(ConsumeUploadQos::getId).collect(Collectors.toList());

        List<Long> uIds = consumeUploadQos.stream().map(ConsumeUploadQos::getUserId).distinct().collect(Collectors.toList());
        List<User> users =null;
        if (uIds.size()!=0) {
            QueryWrapper<User> userWrapper = new QueryWrapper<User>().in("id", uIds);
            users = userMapper.selectList(userWrapper);
        }

        Server mqttBroker = Singleton.INSTANCE.getMqttBroker();

        //没有信息就不发布
        if(!consumeUploadQos.isEmpty()){
            int[] count = {0};
            if (users!=null){
                users.forEach(user ->{
                    MqttPublishMessage message = MqttMessageBuilders.publish()
                            .topicName(user.getTopic())
                            .retained(true)
                            .qos(MqttQoS.EXACTLY_ONCE)
                            .payload(Unpooled.copiedBuffer(JSONUtil.toJsonStr(consumeUploadQos.get(count[0])).getBytes(UTF_8)))
                            .build();
                    //发布数据
                    mqttBroker.internalPublish(message, "INTRLPUB");
                    count[0]++;
                });
            }
        }

        //修改已经发布的数据字段
        if (ids.size()!=0) {
            consumeUploadQosMapper.updatePushByIds(ids);
        }
    }

    @Override
    public String getCron() {
        Cron cron = cronMapper.selectOne(null);
            return cron.getCron();

    }

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        super.configureTasks(taskRegistrar);
    }
}
